package com.shujia.flink.state;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo03CheckPoint {
    public static void main(String[] args) throws Exception {
        // CheckPoint ：将状态定期按策略保存到可靠的存储系统中

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 开启CheckPoint：每 10000ms 开始一次 checkpoint
        // 时间越大导致故障恢复的代价变大，时间越小CK所消耗的资源越多
        // 一般设置为所能容忍的数据延迟大小
        env.enableCheckpointing(10000);
        // 高级选项：
        // 设置模式为精确一次 (这是默认值)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 两次checkpoints 之间的最小延迟
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 允许两个连续的 checkpoint 错误
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        // 同一时间只允许一个 checkpoint 进行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //将状态保存到hdfs中
        env.getCheckpointConfig().setCheckpointStorage("hdfs://master:9000/flink/checkpoint");


        env.execute();

    }
}
